package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo23CheckPoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo23CheckPoint")

    val sc: SparkContext = new SparkContext(conf)
    // 如果代码中有CheckPoint操作，则需要指定CheckPoint目录
    sc.setCheckpointDir("spark/data/ck")

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")

    val newStuRDD: RDD[String] = stuRDD.map(line => {
      // 通过打印的次数判断数据有没有被Spark缓存起来
      println("进入了map方法")
      line
    })
    // RDD做checkpoint之前可以先cache
    newStuRDD.cache()

    // 对多次使用的RDD做CheckPoint
    newStuRDD.checkpoint()

    // 统计班级人数
    newStuRDD.map(line => (line.split(",")(4), 1)).reduceByKey(_ + _).foreach(println)

    // 统计性别人数
    newStuRDD.map(line => (line.split(",")(3), 1)).reduceByKey(_ + _).foreach(println)

    // 统计年龄分布情况
    newStuRDD.map(line => (line.split(",")(2).toInt, 1)).reduceByKey(_ + _).foreach(println)

    while (true) {

    }

  }

}
